package com.edu.yx.app.App_04_DwsApp.Dws_05_Exam;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.ExamScoreDistributionBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;

import static com.edu.yx.common.Constant.*;

public class Dws_03_Exam_Score_Distribution_Count extends BaseApp {

    public static void main(String[] args) {
        new Dws_03_Exam_Score_Distribution_Count().init(
                4053,
                2,
                "Dws_03_Exam_Score_Distribution_Count",
                TOPIC_DWD_EXAM_PAPER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.封装pojo
        //假设某个人考试，在90-100分段考了10次，那么这个人只能算考一次，用一个set去重，最后根据set的长度获取人数
        //由于粒度是试卷+成绩
        SingleOutputStreamOperator<ExamScoreDistributionBean> parseToPojoStream = parseToPojo(stream);

        //2.开窗聚合
        SingleOutputStreamOperator<ExamScoreDistributionBean> windowAndAggStream = windowAndAgg(parseToPojoStream);

        //3.写入doris
        writeToDoris(windowAndAggStream);


    }

    private void writeToDoris(SingleOutputStreamOperator<ExamScoreDistributionBean> windowAndAggStream) {

        windowAndAggStream
                .map(bean ->{

                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);

                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_exam_score_distribution_window"));

    }

    private SingleOutputStreamOperator<ExamScoreDistributionBean> windowAndAgg(SingleOutputStreamOperator<ExamScoreDistributionBean> parseToPojoStream) {

        return parseToPojoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ExamScoreDistributionBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts)-> bean.getTs())
                )
                .keyBy(bean->bean.getPaperId() + "_" + bean.getScoreDistribution())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ExamScoreDistributionBean>() {
                            @Override
                            public ExamScoreDistributionBean reduce(ExamScoreDistributionBean value1,
                                                                    ExamScoreDistributionBean value2) throws Exception {

                                value1.getPersonSet().addAll(value2.getPersonSet());

                                return value1;
                            }
                        }, new ProcessWindowFunction<ExamScoreDistributionBean, ExamScoreDistributionBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<ExamScoreDistributionBean> elements,
                                                Collector<ExamScoreDistributionBean> out) throws Exception {

                                ExamScoreDistributionBean bean = elements.iterator().next();
                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                bean.setPersonCt((long) bean.getPersonSet().size());

                                out.collect(bean);
                            }
                        }
                );
    }

    private SingleOutputStreamOperator<ExamScoreDistributionBean> parseToPojo(DataStreamSource<String> stream) {

       return stream
                .map(strjson-> {

                    JSONObject jsonObject = JSON.parseObject(strjson);

                    String id = jsonObject.getString("id");
                    String scoreDistribution = jsonObject.getString("flag");
                    String userId = jsonObject.getString("user_id");
                    Long ts = jsonObject.getLong("ts")*1000L;

                    return new ExamScoreDistributionBean(

                            "",
                            "",
                            scoreDistribution,
                            id,
                            "",
                            new HashSet<>(Collections.singleton(userId)),
                            0L,
                            ts
                    );
                });

    }


}
